package com.lxb.demo3

import java.time.Instant

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{Schema, Table}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

/**
 * @author: albert
 * @date: 2021/5/27 15:55
 * @description:
 */
object FromDataStreamTest1 {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    val dataStream: DataStream[User] = env.fromElements(User("Alice", 4, Instant.ofEpochMilli(1000)),
      User("Bob", 6, Instant.ofEpochMilli(1001)),
      User("Alice", 10, Instant.ofEpochMilli(1002)))

    // 1 导出所有物理列
//    val inputTable: Table = tableEnv.fromDataStream(dataStream)
//    inputTable.printSchema()

    /**
     * 2.导出物理列并且添加计算列，这个例子主要是添加 proctime 的属性列
     */
//    tableEnv.fromDataStream(
//      dataStream,
//      Schema.newBuilder()
//    .columnByExpression("proc_time11","PROCTIME()")
//    .build()).printSchema()


    /**
     * 3.导出物理列并且添加计算列，该例子是添加rowtime属性，并且添加自定义水印
     */
//    tableEnv.fromDataStream(
//      dataStream,
//      Schema.newBuilder()
//        .columnByExpression("rowtime","CAST(event_time AS TIMESTAMP_LTZ(3))")
//        .watermark("rowtime","rowtime - INTERVAL '10' SECOND")
//        .build()
//    ).printSchema()

    /**
     * 4.导出物理列并且添加计算列 访问流记录的时间戳以创建rowtime属性列也依赖于DataStream API中生成的水印
     * 前提：dataStream已经设置了水印
     */
//    tableEnv.fromDataStream(
//      dataStream,
//      Schema.newBuilder()
//        .columnByMetadata("rowtime","TIMESTAMP_LTZ(3)")
//        .watermark("rowtime", "SOURCE_WATERMARK()")
//        .build()
//    ).printSchema()

    /**
     * 5.导出物理列
     * 我们可以将时间戳的默认精度从9降低到3
     * 我们还对列进行投影并将`event_time`放在开头
     */
    val table: Table =
      tableEnv.fromDataStream(
        dataStream,
        Schema.newBuilder()
          .column("event_time", "TIMESTAMP_LTZ(3)")
          .column("name", "STRING")
          .column("score", "INT")
          .watermark("event_time", "SOURCE_WATERMARK()")
          .build())
    table.printSchema()

  }
}
case class User(name: String, score: java.lang.Integer, event_time: java.time.Instant)